package cn.edu.flink.scala.tutorial.state.keyedstate

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

import scala.Console.in

object ListStateTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val range1 = (1 to 15).toList
    val range2 = (15 to 10 by -1).toList
    val value = env.fromCollection(range1 ::: range2)

    value
      .keyBy(_ => true)
      .map(new MyMap)
      .print()

    env.execute("值状态")
  }


  class MyMap extends RichMapFunction[Int, (String, Int)] {
    // 定义列表状态
    var listState: ListState[Int] = _


    override def open(parameters: Configuration): Unit = {
      listState = getRuntimeContext.getListState(new ListStateDescriptor[Int]("list-state", classOf[Int]))
    }

    override def map(value: Int): (String, Int) = {
      val ints = listState.get().iterator()
      while (ints.hasNext){
        val i = ints.next()
        if (i == value) return (value.toString, 1)
      }
      listState.add(value)
      (value.toString, 0)
    }
  }

}
